package com.reactim.message.service.impl;

import com.reactim.message.dto.MessageResponse;
import com.reactim.message.service.MessagePushService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.List;
import java.util.Set;

/**
 * 消息推送服务实现类
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class MessagePushServiceImpl implements MessagePushService {

    private final ReactiveRedisTemplate<String, Object> redisTemplate;
    private final KafkaTemplate<String, Object> kafkaTemplate;

    // Redis Key 前缀
    private static final String USER_ONLINE_KEY = "user:online:";
    private static final String USER_CONNECTIONS_KEY = "user:connections:";

    // Kafka Topic
    private static final String MESSAGE_PUSH_TOPIC = "message-push";
    private static final String OFFLINE_NOTIFICATION_TOPIC = "offline-notification";

    @Override
    public Mono<Void> pushToUser(Long userId, MessageResponse message) {
        return isUserOnline(userId)
                .flatMap(isOnline -> {
                    if (isOnline) {
                        // 用户在线，通过RSocket推送
                        return pushMessageViaRSocket(userId, message)
                                .doOnSuccess(v -> log.debug("消息推送成功: {} -> {}", message.getMessageId(), userId))
                                .onErrorResume(error -> {
                                    log.warn("RSocket推送失败，转为离线处理: {}", error.getMessage());
                                    return sendOfflineNotification(userId, message);
                                });
                    } else {
                        // 用户离线，发送离线通知
                        return sendOfflineNotification(userId, message);
                    }
                });
    }

    @Override
    public Mono<Void> pushToGroup(Long groupId, MessageResponse message, Long excludeUserId) {
        // 获取群组成员列表（这里需要调用群组服务）
        return getGroupMembers(groupId)
                .filter(memberId -> !memberId.equals(excludeUserId))
                .flatMap(memberId -> pushToUser(memberId, message))
                .then()
                .doOnSuccess(v -> log.info("群组消息推送完成: group={}, message={}", groupId, message.getMessageId()));
    }

    @Override
    public Mono<Void> pushToUsers(List<Long> userIds, MessageResponse message) {
        return Flux.fromIterable(userIds)
                .flatMap(userId -> pushToUser(userId, message))
                .then();
    }

    @Override
    public Mono<Boolean> isUserOnline(Long userId) {
        String key = USER_ONLINE_KEY + userId;
        return redisTemplate.hasKey(key)
                .defaultIfEmpty(false);
    }

    @Override
    public Mono<Integer> getUserConnectionCount(Long userId) {
        String key = USER_CONNECTIONS_KEY + userId;
        return redisTemplate.opsForSet().size(key)
                .map(Long::intValue)
                .defaultIfEmpty(0);
    }

    @Override
    public Mono<Void> sendOfflineNotification(Long userId, MessageResponse message) {
        // 构建离线通知消息
        OfflineNotification notification = OfflineNotification.builder()
                .userId(userId)
                .messageId(message.getMessageId())
                .fromUserId(message.getFromUserId())
                .messageType(message.getMessageType())
                .content(generateNotificationContent(message))
                .timestamp(message.getCreatedAt())
                .build();

        // 发送到Kafka进行异步处理
        return Mono.fromRunnable(() -> {
            kafkaTemplate.send(OFFLINE_NOTIFICATION_TOPIC, userId.toString(), notification);
            log.debug("离线通知已发送: userId={}, messageId={}", userId, message.getMessageId());
        });
    }

    /**
     * 通过RSocket推送消息
     */
    private Mono<Void> pushMessageViaRSocket(Long userId, MessageResponse message) {
        // 构建推送消息
        PushMessage pushMessage = PushMessage.builder()
                .type(PushMessage.PushType.NEW_MESSAGE) // 推送消息类型
                .userId(userId)
                .data(message)
                .timestamp(System.currentTimeMillis())
                .build();

        // 发送到Kafka，由网关服务处理RSocket推送
        return Mono.fromRunnable(() -> {
            kafkaTemplate.send(MESSAGE_PUSH_TOPIC, userId.toString(), pushMessage);
            log.debug("RSocket推送消息已发送到队列: userId={}, messageId={}", userId, message.getMessageId());
        });
    }

    /**
     * 获取群组成员列表
     * TODO: 这里需要调用群组服务的API
     */
    private Flux<Long> getGroupMembers(Long groupId) {
        // 临时实现，实际应该调用群组服务
        return redisTemplate.opsForSet().members("group:members:" + groupId)
                .cast(Long.class)
                .switchIfEmpty(Flux.empty());
    }

    /**
     * 生成通知内容
     */
    private String generateNotificationContent(MessageResponse message) {
        switch (message.getMessageType()) {
            case TEXT:
                return message.getContent().getText() != null ?
                       message.getContent().getText().getText() : "[文本消息]";
            case IMAGE:
                return "[图片]";
            case VOICE:
                return "[语音]";
            case VIDEO:
                return "[视频]";
            case FILE:
                return "[文件]";
            case LOCATION:
                return "[位置]";
            case EMOJI:
                return "[表情]";
            case CONTACT:
                return "[联系人]";
            case URL_CARD:
                return "[链接]";
            case SYSTEM:
                return message.getContent().getSystem() != null ?
                       message.getContent().getSystem().getContent() : "[系统消息]";
            default:
                return "[消息]";
        }
    }

    /**
     * 推送消息数据结构
     */
    @lombok.Data
    @lombok.Builder
    @lombok.NoArgsConstructor
    @lombok.AllArgsConstructor
    public static class PushMessage {
        private PushType type;
        private Long userId;
        private Object data;
        private Long timestamp;

        public enum PushType {
            NEW_MESSAGE,        // 新消息
            MESSAGE_RECALL,     // 消息撤回
            MESSAGE_DELETE,     // 消息删除
            MESSAGE_EDIT,       // 消息编辑
            READ_RECEIPT,       // 已读回执
            TYPING_INDICATOR,   // 正在输入
            USER_STATUS_CHANGE  // 用户状态变更
        }
    }

    /**
     * 离线通知数据结构
     */
    @lombok.Data
    @lombok.Builder
    @lombok.NoArgsConstructor
    @lombok.AllArgsConstructor
    public static class OfflineNotification {
        private Long userId;
        private String messageId;
        private Long fromUserId;
        private com.reactim.message.entity.Message.MessageType messageType;
        private String content;
        private java.time.LocalDateTime timestamp;
    }
}
